-- Copyright (c) 2024 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- You can use this software according to the terms and conditions of the Mulan PSL v2.
-- You may obtain a copy of Mulan PSL v2 at:
--         http://license.coscl.org.cn/MulanPSL2
-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
-- See the Mulan PSL v2 for more details.

local lu = require 'luaunit'
local defs = require 'unit_manager.class.logic_fw.comm_defs'
local cmn = require 'common'
local fw_mgmt_client = require 'general_hardware.client'
local chip = require 'unit_manager.class.logic_fw.cpld_chip'
local drivers = require 'unit_manager.class.logic_fw.upgrade.drivers_api'
local fw_upgrade = require 'unit_manager.class.logic_fw.upgrade.fw_upgrade'
local valid = require 'unit_manager.class.logic_fw.upgrade.valid'
local factory = require 'factory'
local utils = require 'mc.utils'
local file_sec = require 'utils.file'

local FW_VERSION <const> = '2.3.3'
local SYSTEM_ID <const> = 1
local project_dir = os.getenv('PROJECT_DIR')
local cfg_mutil_cold_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_mutil_cold.cfg'
local cfg_mutil_hot_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_mutil_hot.cfg'
local cfg_single_cold_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_single_cold.cfg'
local cfg_i2c_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/update_i2c.cfg'
local file_path <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/Firmware1'
local cpld_01 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld01.svf'
local valid_01 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid01.svf'
local valid_02 <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid02.svf'
local cpld_vme <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld.vme'
local valid_vme <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/valid.vme'
local bin <const> = project_dir .. '/test/unit/test_data/cpld_upgrade/cpld01.bin'
local release_path_ret
local fw_version_ret
local prepare_ret

local signal = {
    fw_list = {
        [1] = {
            csr = {
                DefaultRoute = 0,
                FirmwareRoute = 4,
                I2CUpgradeChip = {},
                Manufacturer   = "Huawei",
                Name = "BCU_CPLD1",
                SmcChip = {},
                SoftwareId = "CPLD-BC83AMDA",
                UId = "00000001020302031825",
                UpgradeChip = {}
            },
            uid = "00000001020302031825",
            name = "CpuBoard1 CPLD1",
            system_id = SYSTEM_ID,
            component_id = 24,
            component_id_ex = 1,
            get_fw_version = function()
                return FW_VERSION
            end,
            set_lock_status = function()
            end,
            i2c_update_chip_lock = function()
                return true, 0
            end,
            i2c_update_chip_unlock = function()
                return true, 0
            end,
            update_chip_lock = function()
                return true, 0
            end,
            update_chip_unlock = function()
                return true, 0
            end,
            switch_to_firmware_route = function()
            end,
            switch_to_default_route = function()
            end,
            chip_info = {
                SetBypassMode = function()
                    return
                end
            },
            update_chip = {
                -- 无法真正写入数据，打桩
                Write = function()
                end
            },
            smc_chip = {
                Read = function()
                    return '\x03\x00\x01'  -- i2c模式+anlu
                end
            },
            i2c_upgrade_chip = {
                Write = function()
                end
            }
        }
    },
    db = {
        CpldValidating = function()
            return {
                ValidatingCpldFlag = 1,
                save = function()
                end
            }
        end
    }
}

TestLogicalFwUpgrade = {}

function TestLogicalFwUpgrade:setupClass()
    -- 公共函数打桩
    cmn.skynet.sleep = function()
    end
    -- 框架提供的接口执行异常，打桩为os.execute
    utils.secure_tar_unzip = function(file_path, release_path)
        local cmd = string.format('/usr/bin/unzip %s -d %s > /dev/null 2>&1', file_path, release_path)
        os.execute(cmd)
        release_path_ret = release_path
    end

    fw_mgmt_client.UpdateServiceUpdateServicePrepareReply = function(_, _, _, _, version, ret, parameters)
        fw_version_ret = version
        prepare_ret = ret
    end

    fw_mgmt_client.UpdateServiceUpdateServiceProcessReply = function()
    end

    fw_mgmt_client.UpdateServiceUpdateServiceFinishReply = function()
    end
end

function TestLogicalFwUpgrade:teardownClass()
end

local function rm_hpm_package()
    local cmd
    if release_path_ret then
        cmd = string.format('/usr/bin/rm %s*.svf > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
        cmd = string.format('/usr/bin/rm %s*.vme > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
        cmd = string.format('/usr/bin/rm %s*.bin > /dev/null 2>&1', release_path_ret)
        os.execute(cmd)
    end
    cmd = string.format('/usr/bin/rm %s', file_path)
    os.execute(cmd)
end

local function build_hpm_package(cpld, cpld_len, vaild, valid_len)
    local file = file_sec.open_s(cpld, 'w+')
    file:write(string.rep('1', cpld_len))
    file:close()

    if vaild then
        file = file_sec.open_s(vaild, 'w+')
        file:write(string.rep('1', valid_len))
        file:close()
    end
    local cmd = string.format('/usr/bin/zip %s -j %s %s > /dev/null 2>&1', file_path, cpld, vaild)
    os.execute(cmd)
    -- 重命名
    cmd = string.format('/usr/bin/mv %s %s', file_path .. '.zip', file_path)
    os.execute(cmd)
    -- 删掉过程文件
    if vaild then
        cmd = string.format('/usr/bin/rm %s %s', cpld, vaild)
    else
        cmd = string.format('/usr/bin/rm %s', cpld)
    end
    os.execute(cmd)
end

local function mock_mutli_cold_valid()
    drivers.get_cpld_device_info = function()
    end

    chip.check_has_pango = function()
        -- 打桩不包含紫光
        return false
    end
    chip.get_cpld_upgrade_file_id = function()
        return 1
    end

    valid.init_valid_file = function()
    end
end

-- 多厂商、冷升级、正常的HPM包
function TestLogicalFwUpgrade:test_mutli_cold_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- 用例场景打桩
    mock_mutli_cold_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 多厂商、冷升级、异常的HPM包
function TestLogicalFwUpgrade:test_mutli_cold_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_02, 1)

    -- 用例场景打桩
    mock_mutli_cold_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

local function mock_mutli_hot_valid()
    factory.get_obj = function()
        return {
            cpu_boards = {},
            exp_boards = {}
        }
    end
end

-- 多厂商、无感升级、正常的HPM包
function TestLogicalFwUpgrade:test_mutli_hot_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- 用例场景打桩
    mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 多厂商、无感升级、异常的HPM包
function TestLogicalFwUpgrade:test_mutli_hot_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_02, 1)

    -- 用例场景打桩
    mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 单厂商、冷升级、正常的HPM包
function TestLogicalFwUpgrade:test_single_cold_valid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_vme, 1, valid_vme, 1)

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_single_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 单厂商、冷升级、异常的HPM包
function TestLogicalFwUpgrade:test_single_cold_invalid_hpm()
    -- 生成hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_single_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

local function mock_i2c_mode()
    fw_mgmt_client.UpdateServiceUpdateServiceUpdateUpgradeStatus = function()
    end
end

-- i2c升级两小盘背板
function TestLogicalFwUpgrade:test_i2c_mode()
    -- 生成hpm包
    build_hpm_package(bin, 0X60000)

    -- 用例场景打桩
    mock_i2c_mode()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_i2c_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

local function start_ac(version)
    defs.UPGRADING_CPLD = 0
    fw_version_ret = version
end

-- 冷升级后AC，升级成功
function TestLogicalFwUpgrade:test_cold_valid_hpm_and_ac()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    mock_mutli_cold_valid()
    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    start_ac(FW_VERSION)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 0)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 冷升级后热升级，升级成功
function TestLogicalFwUpgrade:test_cold_and_hot_valid_hpm()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    mock_mutli_cold_valid()
    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 生成热升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    mock_mutli_hot_valid()

    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_hot_path)
    fw_upgrade.process_upgrade(signal, SYSTEM_ID, _, file_path)
    fw_upgrade.finish_upgrade(signal, SYSTEM_ID, _)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
    lu.assertEquals(defs.UPGRADING_CPLD, 1)
    lu.assertEquals(fw_upgrade.cpld_valid, false)

    -- 清理hpm包
    rm_hpm_package()
end

-- 冷升级生效过程中触发升级，升级失败
function TestLogicalFwUpgrade:test_active_cpld_and_upgrade()
    -- 生成冷升级hpm包
    build_hpm_package(cpld_01, 1, valid_01, 1)
    -- 用例场景打桩
    mock_mutli_cold_valid()
    -- 模拟生效过程中
    fw_upgrade.cpld_valid = true
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)
    fw_upgrade.cpld_valid = false

    -- 校验结果
    lu.assertEquals(prepare_ret, defs.RET.ERR)

    -- 清理hpm包
    rm_hpm_package()
end

-- 测试单Host机型准备阶段版本号返回实际值
function TestLogicalFwUpgrade:test_singlehost_prepare_version()
    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, SYSTEM_ID, _, cfg_mutil_cold_path)

    -- 校验结果
    lu.assertEquals(fw_version_ret, FW_VERSION)
    lu.assertEquals(prepare_ret, defs.RET.OK)
end

-- 测试HPC机型准备阶段版本号返回""
function TestLogicalFwUpgrade:test_mutlihost_prepare_version()
    -- 执行测试
    fw_upgrade.prepare_upgrade(signal, 255, _, cfg_mutil_cold_path)

    -- 校验结果
    lu.assertEquals(fw_version_ret, "")
    lu.assertEquals(prepare_ret, defs.RET.OK)
end

local function deepCopy(original)
    local copy
    if type(original) == "table" then
        copy = {}
        for key, value in pairs(original) do
            copy[key] = deepCopy(value)
        end
    else
        copy = original
    end
    return copy
end

-- 测试process阶段输入异常值
function TestLogicalFwUpgrade:test_outlier()
    -- 制造一个可以赋异常值的signal的拷贝
    local outlier = deepCopy(signal)
    -- 执行测试
    outlier.fw_list = nil
    local ok = pcall(fw_upgrade.process_upgrade, outlier, 1, _, cfg_mutil_cold_path)
    lu.assertEquals(ok, true)
    -- 恢复
    outlier = deepCopy(signal)
    -- 执行测试
    outlier.cfg_list = nil
    ok = pcall(fw_upgrade.process_upgrade, outlier, 1, _, cfg_mutil_cold_path)
    lu.assertEquals(ok, true)
    -- 执行测试file_path为空
    outlier = deepCopy(signal)
    ok = pcall(fw_upgrade.process_upgrade, outlier, 1, _, "")
    lu.assertEquals(ok, true)
end